package util;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.setting.SettingUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mysql.cj.protocol.ResultsetRowsOwner;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;

import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;


public class ClickHouseSinkFunction extends RichSinkFunction {

    public static List<String> cloumnList = ListUtil.toList("key", "value", "values", "like", "table", "view", "date", "datetime");
    private Connection connection;
    private String database;
    private String table;
    private String primaryKey;
    private boolean needReplace;

    public ClickHouseSinkFunction() {

    }


    public ClickHouseSinkFunction(String database, String table, String primaryKey, boolean needReplace) {
        this.database = database;
        this.table = table;
        this.needReplace = needReplace;
        this.primaryKey = primaryKey;
    }

    @Override
    public void open(Configuration parameters) throws Exception {

        Class.forName(GmallConfig.CLICKHOUSE_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.CLICKHOUSE_URL);
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(Object value, Context context) throws Exception {
        PreparedStatement preparedStatement = null;
        try {
            checkSql(value);
            buildDataBase();
            buildTable(value);
            String upsertSql = buildInsertSql(value);
            System.out.println(upsertSql);
            preparedStatement = connection.prepareStatement(upsertSql);
            preparedStatement.executeUpdate();

        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }

    }

    private void checkSql(Object value) {
        if (value instanceof String){
            throw new RuntimeException(" must not be String! allowed POJO,Row,JSONObject ");
        }

        if (value instanceof Row) {
            Row value1 = (Row) value;
            Set<String> fieldNames = value1.getFieldNames(true);
            long count = fieldNames.stream().filter(t -> t.startsWith("EXPR")).count();
            if (count > 0) {
                System.out.println("select sql must use `as` ");
                throw new RuntimeException("select sql must use `as` ");
            }
        }
    }

    private void buildDataBase() {
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement("CREATE DATABASE IF NOT EXISTS  " + database);
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("clickhouse database " + database + " create fail！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private String buildInsertSql(Object row) {

        JSONObject newData  = getJSONObjectValue(row);

        return "insert into " + database + "." + table + "( " +
                StringUtils.join(newData.keySet(), ",") + ") values( '" +
                StringUtils.join(newData.values(), "','") + "')";
    }

    private JSONObject getJSONObjectValue(Object row) {

        if (row instanceof Row){
            JSONObject newData = new JSONObject();
            Row row1 = (Row) row;
            Set<String> fieldNames = row1.getFieldNames(true);
            for (String fieldName : fieldNames) {
                Object field = row1.getField(fieldName);
                newData.put(fieldName, field);
            }
            return newData;
        }
        if (row instanceof JSONObject){
            return (JSONObject) row;
        }
        return JSON.parseObject(JSON.toJSONString(row));
    }

    private void buildTable(Object row) {

        PreparedStatement preparedStatement = null;


        Set<String> fieldNames = getFiledName(row);
//        Set<String> fieldNames = row.getFieldNames(true);

        try {
            StringBuffer createTableSQL = new StringBuffer("create table if not exists ")
                    .append(database)
                    .append(".")
                    .append(table)
                    .append("( ");

            String[] fields = fieldNames.toArray(new String[0]);
            for (int i = 0; i < fieldNames.size(); i++) {
                String field = fields[i];
                createTableSQL.append(field).append(" varchar ");

                if (i < fields.length - 1) {
                    createTableSQL.append(",");
                }
            }

            if (needReplace) {
                createTableSQL.append(")").append("ENGINE = ReplacingMergeTree()\n" +
                        "PRIMARY KEY (" + primaryKey + ") ");
            } else {
                createTableSQL.append(")").append("ENGINE = MergeTree()\n" +
                        "PRIMARY KEY (" + primaryKey + ") ");
            }
            System.out.println(createTableSQL);
            preparedStatement = connection.prepareStatement(createTableSQL.toString());
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("clickhouse table " + table + " create fail ！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private Set<String> getFiledName(Object row) {

        if (row instanceof Row) {
            Row row1 = (Row) row;
            return row1.getFieldNames(true);
        }

        if (row instanceof JSONObject) {
            JSONObject row1 = (JSONObject) row;
            return row1.keySet();
        }

        Field[] fields = row.getClass().getDeclaredFields();

        Set<String> set = new HashSet<>();
        for (Field field : fields) {
            set.add(field.getName());
        }

        return set;
    }


}
